package com.peng.rpc.netty.client;

import com.peng.rpc.entity.RpcRequest;
import com.peng.rpc.entity.RpcResponse;
import com.peng.rpc.netty.util.UnprocessedRequests;
import com.peng.rpc.serializer.CommonSerializer;
import com.peng.rpc.util.SingletonFactory;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.AttributeKey;
import io.netty.util.ReferenceCountUtil;
import lombok.extern.slf4j.Slf4j;

import java.net.InetAddress;
import java.net.InetSocketAddress;

/**
 * @author Wonderson
 * @email wpp.ds@foxmail.com
 * @date 2022/3/1 21:55
 */
@Slf4j
public class NettyClientHandler extends SimpleChannelInboundHandler<RpcResponse> {

    private final UnprocessedRequests unprocessedRequests;

    public NettyClientHandler() {
        unprocessedRequests = SingletonFactory.getSingletonInstance(UnprocessedRequests.class);
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, RpcResponse rpcResponse) throws Exception {
        try {
            log.info("开始解析服务器的响应:{}", rpcResponse);
//            //在各个handler之间设置一个共享的map，key为responseKey，value为此rpcResponse对象
//            AttributeKey<RpcResponse> responseKey = AttributeKey.valueOf("rpcResponse-" + rpcResponse.getRequestId());
//            ctx.channel().attr(responseKey).set(rpcResponse);
//            ctx.channel().close();
            unprocessedRequests.complete(rpcResponse);
        } finally {
            ReferenceCountUtil.release(rpcResponse);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.error("解析响应出现错误：{}", cause.getMessage());
        ctx.close();
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        //抓取空闲连接事件进行处理
        if (evt instanceof IdleStateEvent) {
            IdleState state = ((IdleStateEvent) evt).state();
            if (state == IdleState.WRITER_IDLE) {
                log.info("开始发送心跳包给{}...", ctx.channel().remoteAddress());
                //TODO: to support custom Serializer
                ChannelProvider.getChannel((InetSocketAddress) ctx.channel().remoteAddress(),
                        CommonSerializer.getSerializerByCode(CommonSerializer.PROTOBUF_SERIALIZER));
                RpcRequest heartbeatRequest = new RpcRequest();
                heartbeatRequest.setIsHeartBeat(true);
                ctx.writeAndFlush(heartbeatRequest).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
            }
        } else { //正常包
            super.userEventTriggered(ctx, evt);
        }
    }
}
